package com.group1.realtime.function;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.util.DimUtil;
import com.group1.realtime.util.DruidDSUtil;
import com.group1.realtime.util.RedisUtil;
import com.group1.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.lang.reflect.InvocationTargetException;
import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.Executor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimFunction<T> {

    private Executor threadPool;
    private DruidDataSource dataSource;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 1. 先获取线程池, 线程交给线程池执行
        threadPool = ThreadPoolUtil.getThreadPool();
        // 2. phoenix的连接池
        dataSource = DruidDSUtil.createDataSource();

    }


    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        // 使用多线程+多客户端的方式去连接redis和Phoenix
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    DruidPooledConnection phoenixConn = dataSource.getConnection();
                    Jedis redisClient = RedisUtil.getRedisClient();

                    // 让子类告诉我们要查询的那个表名和id
                    JSONObject dim = DimUtil.readDim(redisClient, phoenixConn, getTable(), getId(input));
                    // 从dim中获取维度, 给T对象中的属性赋值
                    addDim(input, dim);
                    phoenixConn.close();
                    redisClient.close();

                    resultFuture.complete(Collections.singletonList(input));
                } catch (SQLException | IllegalAccessException | InstantiationException | InvocationTargetException e) {

                    e.printStackTrace();
                }
            }
        });
    }
}
